package com.mini.grpc.transport.netty;

import com.mini.grpc.common.Status;
import com.mini.grpc.transport.ServerListener;
import com.mini.grpc.transport.ServerTransport;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http2.Http2SecurityUtil;
import io.netty.handler.codec.http2.Http2Settings;
import io.netty.handler.ssl.ApplicationProtocolConfig;
import io.netty.handler.ssl.ApplicationProtocolNames;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.SupportedCipherSuiteFilter;
import io.netty.handler.ssl.util.SelfSignedCertificate;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.net.ssl.SSLException;
import java.security.cert.CertificateException;

/**
 * NettyServerTransport实现了基于Netty的服务端传输。
 */
public class NettyServerTransport implements ServerTransport {
    
    private static final Logger logger = LoggerFactory.getLogger(NettyServerTransport.class);
    
    private final SocketAddress address;
    private final ServerListener listener;
    private final boolean useSSL;
    
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private Channel channel;
    private final AtomicBoolean started = new AtomicBoolean(false);
    private final AtomicBoolean shutdown = new AtomicBoolean(false);
    private final CountDownLatch terminationLatch = new CountDownLatch(1);
    private int port = -1;
    
    /**
     * 创建一个新的NettyServerTransport
     *
     * @param address 服务端监听地址
     * @param listener 服务端监听器
     * @param useSSL 是否使用SSL
     */
    public NettyServerTransport(SocketAddress address, ServerListener listener, boolean useSSL) {
        this.address = address;
        this.listener = listener;
        this.useSSL = useSSL;
    }
    
    @Override
    public boolean start() {
        if (!started.compareAndSet(false, true)) {
            return false;
        }
        
        try {
            bossGroup = new NioEventLoopGroup(1);
            workerGroup = new NioEventLoopGroup();
            
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        if (useSSL) {
                            SelfSignedCertificate ssc = new SelfSignedCertificate();
                            SslContext sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey())
                                .ciphers(Http2SecurityUtil.CIPHERS, SupportedCipherSuiteFilter.INSTANCE)
                                .applicationProtocolConfig(new ApplicationProtocolConfig(
                                    ApplicationProtocolConfig.Protocol.ALPN,
                                    ApplicationProtocolConfig.SelectorFailureBehavior.NO_ADVERTISE,
                                    ApplicationProtocolConfig.SelectedListenerFailureBehavior.ACCEPT,
                                    ApplicationProtocolNames.HTTP_2))
                                .build();
                            ch.pipeline().addLast(sslCtx.newHandler(ch.alloc()));
                        }
                        
                        // 配置HTTP/2处理器
                        ch.pipeline().addLast(new Http2ServerInitializer(listener));
                    }
                });
            
            // 绑定端口并启动服务器
            ChannelFuture f = b.bind(address).sync();
            channel = f.channel();
            
            if (address instanceof InetSocketAddress) {
                port = ((InetSocketAddress) channel.localAddress()).getPort();
            }
            
            logger.info("Server started and listening on port {}", port);
            
            return true;
        } catch (Exception e) {
            logger.error("Failed to start server", e);
            shutdown(Status.INTERNAL.withDescription("Failed to start server: " + e.getMessage()));
            return false;
        }
    }
    
    @Override
    public void shutdown(Status status) {
        if (!shutdown.compareAndSet(false, true)) {
            return;
        }
        
        logger.info("Shutting down server: {}", status);
        
        try {
            if (channel != null) {
                channel.close().sync();
            }
            
            if (bossGroup != null) {
                bossGroup.shutdownGracefully(0, 5, TimeUnit.SECONDS);
            }
            
            if (workerGroup != null) {
                workerGroup.shutdownGracefully(0, 5, TimeUnit.SECONDS);
            }
            
            listener.transportTerminated();
        } catch (Exception e) {
            logger.error("Error during server shutdown", e);
        } finally {
            terminationLatch.countDown();
        }
    }
    
    @Override
    public int getPort() {
        return port;
    }
    
    @Override
    public boolean isShutdown() {
        return shutdown.get();
    }
    
    @Override
    public boolean isTerminated() {
        return terminationLatch.getCount() == 0;
    }
    
    @Override
    public void awaitTermination() throws InterruptedException {
        terminationLatch.await();
    }
} 